
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types._
import org.apache.spark.sql.{DataFrame, Row, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by yangqiyuan on 2018/3/26.
  */
object DmpApp {

  def toIntE(name:String):Int={
    try{
    name.toInt}
    catch {
      case e:Exception=>0
    }

  }


  def main(args: Array[String]): Unit = {

    val conf = new SparkConf()

      .setMaster("local")
      .setAppName(DmpApp.getClass.getSimpleName)
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .set("spark.sql.parquet.compression.codec", "snappy")
    val spark = new SparkContext(conf)
    val sqlContext = new SQLContext(spark)
    val file: RDD[String] = spark.textFile("ad.FINISH")
    val rdd: RDD[Row] =
    file.map(x=> x.split(",",-1)).filter(_.length>=85).map(split => {



     Row(split(0).toString, split(1).toInt, split(2).toInt, split(3).toInt, split(4).toInt, split(5).toString, split(6).toString, split(7).toInt, split(8).toInt, split(9).toDouble,
        split(10).toDouble, split(11).toString, split(12).toString, split(13).toString, split(14).toString, split(15).toString, split(16).toString, split(17).toInt, split(18).toString,
        split(19).toString, split(20).toString, split(21).toString, split(22).toString, split(23).toString, split(24).toString, split(25).toString, split(26).toInt, split(27).toString, split(28).toInt,
        split(29).toString, split(30).toInt, split(31).toInt, split(32).toInt, split(33).toString, split(34).toInt, split(35).toInt, split(36).toInt, split(37).toString, split(38).toInt, split(39).toInt,
        split(40).toDouble, split(41).toDouble, split(42).toInt, split(43).toString, split(44).toDouble, split(45).toDouble, split(46).toString, split(47).toString, split(48).toString,
        split(49).toString, split(50).toString, split(51).toString, split(52).toString, split(53).toString, split(54).toString, split(55).toString, split(56).toString, split(57).toInt, split(58).toDouble,
        split(59).toString, split(60).toString, split(61).toString, split(62).toString, split(63).toString, split(64).toString, split(65).toString, split(66).toString, split(67).toString, split(68).toString,
        split(69).toString,
        split(70).toString,
        split(71).toString,
        split(72).toString,
        toIntE(split(73)),
        split(74).toDouble,
        split(75).toDouble,
       split(76).toDouble,
       split(77).toDouble,
       split(78).toDouble,
       split(79).toString,
       split(80).toString,
       split(81).toString,
       split(82).toString,
       split(83).toString,
       toIntE(split(84)))

    })
    val structType: StructType = StructType(
      List(
        StructField("sessionid", StringType),
        StructField("advertisersid", IntegerType),
        StructField("adorderid", IntegerType),
        StructField("adcreativeid", IntegerType),
        StructField("adplatformprov", IntegerType),
        StructField("sdkversion", StringType),
        StructField("adplatformkey", StringType),
        StructField("putinmodeltype", IntegerType),
        StructField("requestmode", IntegerType),
        StructField("adprice", DoubleType),
        StructField("adppprice", DoubleType),
        StructField("requestdate", StringType),
        StructField("ip", StringType),
        StructField("appid", StringType),
        StructField("appname", StringType),// -------------------------------------------
        StructField("uuid", StringType),
        StructField("device", StringType),
        StructField("client", IntegerType),
        StructField("osversion", StringType),
        StructField("density", StringType),
        StructField("pw", StringType),
        StructField("ph", StringType),
        StructField("long", StringType),//23
        StructField("lat", StringType),
        StructField("provincename", StringType),
        StructField("cityname", StringType),//26
        StructField("ispid", IntegerType),
        StructField("ispname", StringType),
        StructField("networkmanneri", IntegerType),//29
        StructField("networkmannern", StringType),
        StructField("iseffective", IntegerType),
        StructField("isbilling", IntegerType),
        StructField("adspacetype", IntegerType),//33
        StructField("adspacetypenam", StringType),
        StructField("devicetype", IntegerType),
        StructField("processnode", IntegerType),
        StructField("apptype", IntegerType),//37
        StructField("district", StringType),
        StructField("paymode", IntegerType),
        StructField("isbid", IntegerType),//40
        StructField("bidprice", DoubleType),
        StructField("winprice", DoubleType),
        StructField("iswin", IntegerType),//43
        StructField("cur", StringType),
        StructField("rate", DoubleType),
        StructField("cnywinprice", DoubleType),//46
        StructField("imei", StringType),
        StructField("mac", StringType),
        StructField("idfa", StringType),
        StructField("openudid", StringType),
        StructField("androidid", StringType),
        StructField("rtbprovince", StringType),
        StructField("rtbcity", StringType),
        StructField("rtbdistrict", StringType),
        StructField("rtbstreet", StringType),
        StructField("storeurl", StringType),
        StructField("realip", StringType),//57
        StructField("isqualityapp", IntegerType),
        StructField("bidfloor", DoubleType),//59
        StructField("aw", StringType),
        StructField("ah", StringType),
        StructField("imeimd5", StringType),
        StructField("macmd5", StringType),
        StructField("idfamd5", StringType),
        StructField("openudidmd5", StringType),
        StructField("androididmd5", StringType),
        StructField("imeisha1", StringType),
        StructField("macsha1", StringType),
        StructField("idfasha1", StringType),
        StructField("openudidsha1", StringType),
        StructField("androididsha1", StringType),
        StructField("uuidunknow", StringType),
        StructField("userid", StringType),//73
        StructField("iptype", IntegerType),
        StructField("initbidprice", DoubleType),//没问题
        StructField("adpayment", DoubleType),
        StructField("agentrate", DoubleType),
        StructField("lomarkrate", DoubleType),
        StructField("adxrate", DoubleType),//79
        //79
        StructField("title", StringType),//没问题
        StructField("keywords", StringType),
        StructField("tagid", StringType),
        StructField("callbackdate", StringType),
        StructField("channelid", StringType),
        StructField("mediatype", IntegerType)

      ))
    val createDataFrame: DataFrame = sqlContext.createDataFrame(rdd, structType)
    createDataFrame.registerTempTable("dmp")
    val sql: DataFrame = sqlContext.sql("select * from dmp")
    sql.show(100)
   // createDataFrame.write.save("parquet")


    //过滤出字段数量等于85的


    // 导入隐式转换

    /*  val dF = res_filter.toDS("sessionid","advertisersid:","adorderid","adcreativeid","adplatformproviderid","sdkversion",
        "adplatformkey","putinmodeltype","requestmode","adprice","adppprice","requestdate","ip","appid"
        ,"appname","uuid","",
        "device","client","osversion","density","pw","ph","long","lat","provincename","cityname","ispid"
        ,"ispname","networkmannerid","networkmannername","iseffective","isbilling","adspacetype","adspacetypename",
        "devicetype","processnode","apptype","district","paymode",
        "isbid","bidprice","winprice","iswin","cur","rate","cnywinprice","imei","mac","idfa","openudid",
        "androidid","rtbprovince","rtbcity","rtbdistrict","rtbstreet","storeurl","realip","isqualityapp",
        "bidfloor","aw","ah","imeimd5",
        "macmd5","idfamd5","openudidmd5","androididmd5","imeisha1","macsha1","idfasha1","openudidsha1",
        "androididsha1","uuidunknow","userid","iptype","initbidprice","adpayment","agentrate","lomarkrate",
        "adxrate","title:","keywords","tagid","callbackdate","channelid","mediatype")
      dF.write.save("parquet")*/

    spark.stop()


  }
}

/*
case class Dmp(sessionid: String,
               advertisersid: Int,
               adorderid: Int,
               adcreativeid: Int,
               adplatformproviderid: Int,
               sdkversion: String,
               adplatformkey: String,
               putinmodeltype: Int,
               requestmode: Int,
               adprice: Double,
               adppprice: Double,
               requestdate: String,
               ip: String,
               appid: String,
               appname: String,
               uuid: String,
               device: String,
               client: Int,
               osversion: String,
               density: String,
               pw: Int,
               ph: Int,
               long: String,
               lat: String,
               provincename: String,
               cityname: String,
               ispid: Int,
               ispname: String,
               networkmannerid: Int,
               networkmannername: String,
               ISBILLING: Int,
               isbilling: Int,
               adspacetype: Int,
               adspacetypename: String, //34
               devicetype: Int,
               processnode: Int,
               apptype: Int,
               district: String,
               paymode: Int,
               isbid: Int,
               bidprice: Double,
               winprice: Double,
               iswin: Int, //43
               cur: String,
               rate: Double,
               cnywinprice: Double,
               imei: String, //47
               mac: String,
               idfa: String,
               openudid: String,
               androidid: String,
               rtbprovince: String,
               rtbcity: String,
               rtbdistrict: String,
               rtbstreet: String,
               storeurl: String,
               realip: String, //57
               isqualityapp: Int,
               bidfloor: Double,
               aw: Int,
               ah: Int,
               imeimd5: String, //62
               macmd5: String,
               idfamd5: String,
               openudidmd5: String,
               androididmd5: String,
               imeisha1: String,
               macsha1: String,
               idfasha1: String,
               openudidsha1: String,
               androididsha1: String,
               uuidunknow: String, //72
               //73
               userid: String,
               //76
               iptype: Int,
               initbidprice: Double,
               adpayment: Double,
               agentrate: Double,
               lomarkrate: Double,
               adxrate: Double,
               title: String, //80
               keywords: String,
               tagid: String,
               callbackdate: String,
               channelid: String,
               mediatype: Int
              )*/
